package org.apache.rocketmq.proxy.grpc.v2.route;

import apache.rocketmq.v2.*;
import com.google.common.net.HostAndPort;
import org.apache.rocketmq.common.attribute.TopicMessageType;
import org.apache.rocketmq.common.constant.PermName;
import org.apache.rocketmq.proxy.common.Address;
import org.apache.rocketmq.proxy.common.ProxyContext;
import org.apache.rocketmq.proxy.config.ConfigurationManager;
import org.apache.rocketmq.proxy.grpc.v2.AbstractMessingActivity;
import org.apache.rocketmq.proxy.grpc.v2.channel.GrpcChannelManager;
import org.apache.rocketmq.proxy.grpc.v2.common.GrpcClientSettingsManager;
import org.apache.rocketmq.proxy.grpc.v2.common.ResponseBuilder;
import org.apache.rocketmq.proxy.processor.MessagingProcessor;
import org.apache.rocketmq.proxy.service.route.ProxyTopicRouteData;
import org.apache.rocketmq.remoting.protocol.route.QueueData;

import java.util.*;
import java.util.concurrent.CompletableFuture;


/**
 * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
 * @author：陈清风扬，个人微信号：chenqingfengyangjj。
 * @date:2025/1/3
 * @方法描述：专门处理根据主题查询路由信息的活动处理器
 */
public class RouteActivity extends AbstractMessingActivity {

    //构造方法，在这里把messagingProcessor消息处理器传递给父类了
    public RouteActivity(MessagingProcessor messagingProcessor, GrpcClientSettingsManager grpcClientSettingsManager, GrpcChannelManager grpcChannelManager) {
        super(messagingProcessor, grpcClientSettingsManager, grpcChannelManager);
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/3
     * @方法描述：根据主题查询路由信息的方法，在第四版本代码中，这个方法还是一个伪方法，和源码一点也不一样，等后面版本我会为大家重构成它本来的样子
     */
    public CompletableFuture<QueryRouteResponse> queryRoute(ProxyContext ctx, QueryRouteRequest request) {
        //创建一个CompletableFuture对象，封装查询到的路由信息
        CompletableFuture<QueryRouteResponse> future = new CompletableFuture<>();
        try {
            //验证请求中的主题是否合法
            validateTopic(request.getTopic());
            //这里有一个操作，就是把请求中的地址列表转换成地址列表，首先我们要知道，请求中的Endpoints其实就是Proxy服务的地址列表
            //这里就是得到了Proxy服务的地址列表集合，这个地址列表集合到时会会替换Broker的地址列表集合，因为在5.0之后的MQ中，客户端并不会直接和Broker通信，而是通过Proxy服务来通信
            //所以要把Proxy的地址返回给客户端
            List<org.apache.rocketmq.proxy.common.Address> addressList = this.convertToAddressList(request.getEndpoints());
            //从请求中得到主题名称
            String topicName = request.getTopic().getName();
            //根据主题名称查询路由数据
            ProxyTopicRouteData proxyTopicRouteData = this.messagingProcessor.getTopicRouteDataForProxy(ctx, addressList, topicName);
            //创建一个消息队列列表，用于存放生成的消息队列
            List<MessageQueue> messageQueueList = new ArrayList<>();
            //构建Broker映射关系，其中key为brokerName，内层key为brokerID，value为Broker实例
            Map<String, Map<Long, Broker>> brokerMap = buildBrokerMap(proxyTopicRouteData.getBrokerDatas());
            //得到该主题的消息类型
            TopicMessageType topicMessageType = messagingProcessor.getMetadataService().getTopicMessageType(ctx, topicName);
            //遍历该主题下的所有队列
            for (QueueData queueData : proxyTopicRouteData.getQueueDatas()) {
                String brokerName = queueData.getBrokerName();
                //这里是根据BrokerName得到BrokerIdMap
                Map<Long, Broker> brokerIdMap = brokerMap.get(brokerName);
                if (brokerIdMap == null) {
                    break;
                }
                //遍历BrokerIdMap的value
                for (Broker broker : brokerIdMap.values()) {
                    //为每一个Broker创建消息队列对象列表，然后把消息队列添加到消息队列列表中
                    messageQueueList.addAll(this.genMessageQueueFromQueueData(queueData, request.getTopic(), topicMessageType, broker));
                }
            }//这行带我是我自己添加的，用来验证测试类的正确性
            System.out.println("查询路由信息成功,要把消息队列返回给客户端了"+messageQueueList);
            //创建响应对象
            QueryRouteResponse response = QueryRouteResponse.newBuilder()
                    .setStatus(ResponseBuilder.getInstance().buildStatus(Code.OK, Code.OK.name()))
                    .addAllMessageQueues(messageQueueList)
                    .build();
            //完成响应
            future.complete(response);
        } catch (Throwable t) {
            future.completeExceptionally(t);
        }
        return future;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/22
     * @方法描述：返回一个地址列表集合，这个地址列表集合就是Proxy服务的地址列表集合
     */
    protected List<org.apache.rocketmq.proxy.common.Address> convertToAddressList(Endpoints endpoints) {

        boolean useEndpointPort = ConfigurationManager.getProxyConfig().isUseEndpointPortFromRequest();

        List<org.apache.rocketmq.proxy.common.Address> addressList = new ArrayList<>();
        for (apache.rocketmq.v2.Address address : endpoints.getAddressesList()) {
            int port = ConfigurationManager.getProxyConfig().getGrpcServerPort();
            if (useEndpointPort) {
                port = address.getPort();
            }
            addressList.add(new org.apache.rocketmq.proxy.common.Address(
                    org.apache.rocketmq.proxy.common.Address.AddressScheme.valueOf(endpoints.getScheme().name()),
                    HostAndPort.fromParts(address.getHost(), port)));
        }

        return addressList;
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/22
     * @方法描述：构建Broker映射关系的方法，其中key为brokerName，内层key为brokerID，value为Broker实例
     * 这个方法我其实不想添加注释，因为这个方法其实就是把创建了主题的BrokerData中，每一个Broker主从节点的信息收集到一起了
     * 并且是按照严格的BrokerName和BrokerId的方式收集的，也就是该方法返回值的映射关系收集的。我之所以不想添加注释是因为这个方法逻辑非常简单
     * 大家自己看看肯定也能看明白，还有一个原因就是在5.0之后的MQ中，Broker的地址信息是不会直接返回给客户端的，而是Proxy服务的地址信息会返回给客户端
     * 所以这个方法相当于在做无用功，大家简单看看就行了
     */
    protected Map<String /*brokerName*/, Map<Long /*brokerID*/, Broker>> buildBrokerMap(
            List<ProxyTopicRouteData.ProxyBrokerData> brokerDataList) {
        Map<String, Map<Long, Broker>> brokerMap = new HashMap<>();
        for (ProxyTopicRouteData.ProxyBrokerData brokerData : brokerDataList) {
            Map<Long, Broker> brokerIdMap = new HashMap<>();
            String brokerName = brokerData.getBrokerName();
            //接下来就是得到BrokerData中每一个Broker节点的网络地址，别忘了，这个地址已经被替换成Proxy服务的地址了
            for (Map.Entry<Long, List<org.apache.rocketmq.proxy.common.Address>> entry : brokerData.getBrokerAddrs().entrySet()) {
                //得到Broker节点的Id
                Long brokerId = entry.getKey();
                List<apache.rocketmq.v2.Address> addressList = new ArrayList<>();
                //使用的是IPv4的地址方案
                AddressScheme addressScheme = AddressScheme.IPv4;
                for (org.apache.rocketmq.proxy.common.Address address : entry.getValue()) {
                    addressScheme = AddressScheme.valueOf(address.getAddressScheme().name());
                    //在这里把每一个Broker节点的网络地址收集到一起了，注意，这里的addressList存放的是一个BrokerData中的所有Broker节点的网络地址
                    //也就是一个Broker主从组的所有节点的网络地址
                    addressList.add(apache.rocketmq.v2.Address.newBuilder()
                            .setHost(address.getHostAndPort().getHost())
                            .setPort(address.getHostAndPort().getPort())
                            .build());
                }
                //在这里创建了Broker实例，这里大家可以看到，创建的每一个Broker对象，如果这些Broker对象是属于同一个BrokerData中
                //也就是同一个Broker主从集群中，那么这些Broker对象除了Id不一样，Endpoints都是一样的，都是Proxy服务器的地址
                //大家完全可以自己打断点看一看
                Broker broker = Broker.newBuilder()
                        //设置BrokerName
                        .setName(brokerName)
                        //设置BrokerId
                        .setId(Math.toIntExact(brokerId))
                        //设置Broker的网络地址，其实就是Proxy的网络地址
                        .setEndpoints(Endpoints.newBuilder()
                                .setScheme(addressScheme)
                                .addAllAddresses(addressList)
                                .build())
                        .build();
                brokerIdMap.put(brokerId, broker);
            }
            brokerMap.put(brokerName, brokerIdMap);
        }
        return brokerMap;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/22
     * @方法描述：根据主题队列信息生成消息队列集合的方法
     */
    protected List<MessageQueue> genMessageQueueFromQueueData(QueueData queueData, Resource topic,
                                                              TopicMessageType topicMessageType, Broker broker) {
        //存放该主题消息队列的集合
        List<MessageQueue> messageQueueList = new ArrayList<>();

        //初始化一些变量，用来定义读写队列，只读队列，只写队列的数量
        int r = 0;
        int w = 0;
        int rw = 0;
        //检查该主题下读写消息的权限
        if (PermName.isWriteable(queueData.getPerm()) && PermName.isReadable(queueData.getPerm())) {
            //如果该主题下的消息都是可读写的，那么计算可读写的队列数量
            //我们创建的TestTopic主题下的队列都是可读写的，读队列和写队列数量都是8，所以这里的rw=8
            rw = Math.min(queueData.getWriteQueueNums(), queueData.getReadQueueNums());
            //计算只读队列的数量，这里得到的就是0
            r = queueData.getReadQueueNums() - rw;
            //计算只写队列的数量，这里得到的也是0
            w = queueData.getWriteQueueNums() - rw;
        } else if (PermName.isWriteable(queueData.getPerm())) {
            //如果只有写权限，那么在这里得到写队列的数量
            w = queueData.getWriteQueueNums();
        } else if (PermName.isReadable(queueData.getPerm())) {
            //如果只有读权限，那么在这里得到读队列的数量
            r = queueData.getReadQueueNums();
        }
        //定义一个变量，表示队列Id
        int queueIdIndex = 0;
        //循环创建只读队列
        for (int i = 0; i < r; i++) {
            //创建一个只读权限的消息队列
            MessageQueue messageQueue = MessageQueue.newBuilder().setBroker(broker).setTopic(topic)
                    //设置队列Id
                    .setId(queueIdIndex++)
                    //设置只读权限
                    .setPermission(Permission.READ)
                    //设置主题消息类型
                    .addAllAcceptMessageTypes(parseTopicMessageType(topicMessageType))
                    .build();
            //把创建的消息队列添加到消息队列集合中
            messageQueueList.add(messageQueue);
        }

        //循环创建只写队列
        for (int i = 0; i < w; i++) {
            MessageQueue messageQueue = MessageQueue.newBuilder().setBroker(broker).setTopic(topic)
                    .setId(queueIdIndex++)
                    .setPermission(Permission.WRITE)
                    .addAllAcceptMessageTypes(parseTopicMessageType(topicMessageType))
                    .build();
            messageQueueList.add(messageQueue);
        }

        //循环创建可读写队列
        for (int i = 0; i < rw; i++) {
            MessageQueue messageQueue = MessageQueue.newBuilder().setBroker(broker).setTopic(topic)
                    .setId(queueIdIndex++)
                    .setPermission(Permission.READ_WRITE)
                    .addAllAcceptMessageTypes(parseTopicMessageType(topicMessageType))
                    .build();
            messageQueueList.add(messageQueue);
        }
        return messageQueueList;
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/22
     * @方法描述：解析主题消息类型的方法，根据主题消息类型返回消息类型列表
     */
    private List<MessageType> parseTopicMessageType(TopicMessageType topicMessageType) {
        switch (topicMessageType) {
            case NORMAL:
                return Collections.singletonList(MessageType.NORMAL);
            case FIFO:
                return Collections.singletonList(MessageType.FIFO);
            case TRANSACTION:
                return Collections.singletonList(MessageType.TRANSACTION);
            case DELAY:
                return Collections.singletonList(MessageType.DELAY);
            case MIXED:
                return Arrays.asList(MessageType.NORMAL, MessageType.FIFO, MessageType.DELAY, MessageType.TRANSACTION);
            default:
                return Collections.singletonList(MessageType.MESSAGE_TYPE_UNSPECIFIED);
        }
    }
}
